#ifndef _JOIN_BOX_H_JM_20110904
#define _JOIN_BOX_H_JM_20110904



#include "GenericBoxModel.h"
#include "../table/JoinContext.h"

#include <map>
#include <utility>

namespace jm4bbx {
//
//class JoinBox : public bobox::basic_box2  {
//public:
//
//
//    enum { INPUT_GROUP_1 = 0, INPUT_GROUP_2 = 1, OUTPUT_GROUP = 2 };
//    enum { INPUT_G1_COLUMN_1 = 0, INPUT_G1_COLUMN_2 = 1, INPUT_G2_COLUMN_1 = 2, INPUT_G2_COLUMN_2 = 3,
//            OUTPUT_COLUMN_1 = 4, OUTPUT_COLUMN_2 = 5 };
//    
//    typedef GenericBoxModel<JoinBox, void> model;
//      
//    JoinBox(bobox::box_id_type id, bobox::box_tid_type tid, bobox::request *req, 
//        const bobox::parameters &params) : bobox::basic_box2(id, tid, req), 
//        index1_(0), index2_(0), outputSize_(0) {     
//
//            if (params.has_parameter("outSize")) {
//                params.get_parameter("outSize", outputSize_);
//            } 
//    }
//
//    virtual void init_impl() {
//		add_group(group_index_type(INPUT_GROUP_1), bobox::column_index_type(INPUT_G1_COLUMN_1), 
//            get_input_descriptors(bobox::inarc_index_type(INPUT_G1_COLUMN_1)));
//		
//        add_group(group_index_type(INPUT_GROUP_2), bobox::column_index_type(INPUT_G2_COLUMN_1), 
//            get_input_descriptors(bobox::inarc_index_type(INPUT_G2_COLUMN_1)));
//
//        add_group(
//			group_index_type(OUTPUT_GROUP), bobox::column_index_type(OUTPUT_COLUMN_1), 	
//            get_output_descriptors(bobox::outarc_index_type(OUTPUT_COLUMN_1)));
//
//        wait();
//      }
//
//
//      void wait() {
//        receive(bobox::inarc_index_type(INPUT_G1_COLUMN_1), group_index_type(INPUT_GROUP_1));
//        receive(bobox::inarc_index_type(INPUT_G2_COLUMN_1), group_index_type(INPUT_GROUP_2));
//      }
//
//      virtual void sync_mach_etwas() {          
//          //nestedLoopJoin();
//          hashJoin();
//      }
//
//
//      void hashJoin() {
//        const group& g1 = get_group(group_index_type(INPUT_GROUP_1));
//        const group& g2 = get_group(group_index_type(INPUT_GROUP_2));
//	    if (g1.poisoned_ || g2.poisoned_) {
//            assert(g1.poisoned_ && g2.poisoned_);
//            send_poisoned(bobox::outarc_index_type(OUTPUT_COLUMN_1));
//            wait();
//            return;
//        }
//        outputSize_ = outputSize_ ? outputSize_ : g1.rdend_ * g2.rdend_;
//    
//
//        if (mapa_.empty()) {
//            for (size_t index = 0; index < g1.rdend_; ++index) {
//                const int key = get_data<bobox::int_policy>(bobox::column_index_type(INPUT_G1_COLUMN_1))[index];
//                const int lineIndex = get_data<bobox::int_policy>(bobox::column_index_type(INPUT_G1_COLUMN_2))[index]; 
//                mapa_.insert(std::make_pair(key, lineIndex));
//            }
//        }
//
//        allocate(group_index_type(OUTPUT_GROUP), outputSize_);
//        size_t outputIndex = 0;
//
//
//        for (; index1_ < g2.rdend_; ++index1_) {
//            const int key = get_data<bobox::int_policy>(bobox::column_index_type(INPUT_G2_COLUMN_1))[index1_];
//            const int lineIndex = get_data<bobox::int_policy>(bobox::column_index_type(INPUT_G2_COLUMN_2))[index1_];
//            auto range = mapa_.equal_range(key);
//            //  pokial sa minule spracovanie zastavilo pocas prechodu nejakou 
//            //  range tak pretocim na posledny spracovany prvok
//            auto it = range.first;
//            for (size_t i = 0; i < index2_; ++it, ++i);
//
//            for (; it != range.second; ++it, ++index2_) {
//                get_data<bobox::int_policy>(bobox::column_index_type(OUTPUT_COLUMN_1))[outputIndex] = (*it).second; 
//                get_data<bobox::int_policy>(bobox::column_index_type(OUTPUT_COLUMN_2))[outputIndex] = lineIndex;
//                ++outputIndex;
//                if (outputIndex == outputSize_) {
//                    ++index2_;
//                    get_group(group_index_type(OUTPUT_GROUP)).wrpos_ = outputIndex;
//                    send(bobox::outarc_index_type(OUTPUT_COLUMN_1), group_index_type(OUTPUT_GROUP));
//                    return;
//                }
//            }
//            index2_ = 0;
//        }
//
//
//
//        get_group(group_index_type(OUTPUT_GROUP)).wrpos_ = outputIndex;
//        send(bobox::outarc_index_type(OUTPUT_COLUMN_1), group_index_type(OUTPUT_GROUP));
//        index1_ = index2_ = 0;
//        wait();
//
//      }
//
//      void nestedLoopJoin() {
//        const group& g1 = get_group(group_index_type(INPUT_GROUP_1));
//        const group& g2 = get_group(group_index_type(INPUT_GROUP_2));
//	    if (g1.poisoned_ || g2.poisoned_) {
//            assert(g1.poisoned_ && g2.poisoned_);
//            send_poisoned(bobox::outarc_index_type(OUTPUT_COLUMN_1));
//            wait();
//            return;
//        }
//        
//        outputSize_ = outputSize_ ? outputSize_ : g1.rdend_ * g2.rdend_;
//
//        allocate(group_index_type(OUTPUT_GROUP), outputSize_);
//        size_t outputIndex = 0;
//
//        for (;index1_ < g1.rdend_; ++index1_) {
//            int key1 = get_data<bobox::int_policy>(bobox::column_index_type(INPUT_G1_COLUMN_1))[index1_];
//            for (; index2_ < g2.rdend_; ++index2_) {
//                int key2 = get_data<bobox::int_policy>(bobox::column_index_type(INPUT_G2_COLUMN_1))[index2_];
//                if (key1 == key2) {
//                    get_data<bobox::int_policy>(bobox::column_index_type(OUTPUT_COLUMN_1))[outputIndex] = 
//                        get_data<bobox::int_policy>(bobox::column_index_type(INPUT_G1_COLUMN_2))[index1_]; 
//                    get_data<bobox::int_policy>(bobox::column_index_type(OUTPUT_COLUMN_2))[outputIndex] = 
//                        get_data<bobox::int_policy>(bobox::column_index_type(INPUT_G2_COLUMN_2))[index2_];
//                    ++outputIndex;
//                    if (outputIndex == outputSize_) {
//                        get_group(group_index_type(OUTPUT_GROUP)).wrpos_ = outputIndex;
//                        ++index2_;
//                        send(bobox::outarc_index_type(OUTPUT_COLUMN_1), group_index_type(OUTPUT_GROUP));
//                        return;
//                    }
//                }
//            }
//            index2_ = 0;
//        }
//
//        get_group(group_index_type(OUTPUT_GROUP)).wrpos_ = outputIndex;
//        send(bobox::outarc_index_type(OUTPUT_COLUMN_1), group_index_type(OUTPUT_GROUP));
//        index1_ = index2_ = 0;
//        wait();
//      }
//
//protected:
//    size_t index1_, index2_;
//    std::multimap<int, int> mapa_;
//    int outputSize_;
//};

} //    namespace jm4bbx
#endif
